package com.practice.gmall.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.practice.gmall.realtime.app.BaseApp;
import com.practice.gmall.realtime.bean.UserLoginBean;
import com.practice.gmall.realtime.common.Constant;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.time.Duration;

/**
 * ClassName: DwsUserLoginWindow04
 * Package: com.practice.gmall.realtime.app.dws
 * Description:  dws层用户域登录各窗口汇总表，统计7日回流用户和当日不同用户数
 * @Author lzy.ethan@qq.com
 * @Create 2023-02-18 15:03
 */
public class DwsUserLoginWindow04 extends BaseApp {
    public static void main(String[] args) {
        new DwsUserLoginWindow04().init(
                4005,
                "DwsUserLoginWindow04",
                2,
                Constant.KAFKA_BROKERS,
                Constant.TOPIC_DWD_TRAFFIC_PAGE
        );
    }

    @Override
    protected void handle(StreamExecutionEnvironment env, DataStreamSource<String> stream) {
        // 1. wrap data to pojo
        SingleOutputStreamOperator<UserLoginBean> beanStream = parseToPojo(stream);

        // 2. compute
        computeStatistic(beanStream);

        // 3. write result to clickhouse
    }


    private void computeStatistic(SingleOutputStreamOperator<UserLoginBean> stream) {

    }

    private SingleOutputStreamOperator<UserLoginBean> parseToPojo(DataStreamSource<String> stream) {
        return stream
                .map(JSON::parseObject)
                .keyBy(obj -> obj.getJSONObject("common").getString("uid"))
                .map(new MapFunction<JSONObject, UserLoginBean>() {
                    @Override
                    public UserLoginBean map(JSONObject value) throws Exception {

                        return new UserLoginBean();
                    }
                })
                .assignTimestampsAndWatermarks(WatermarkStrategy
                        .<UserLoginBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                        .withIdleness(Duration.ofSeconds(2))
                        .withTimestampAssigner((element, recordTimestamp) -> element.getTs()));
    }
}
